package com.nageoffer.onecoupon.merchant.admin.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.IdUtil;
import com.alibaba.excel.EasyExcel;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.nageoffer.onecoupon.framework.exception.ClientException;
import com.nageoffer.onecoupon.merchant.admin.common.context.UserContext;
import com.nageoffer.onecoupon.merchant.admin.common.enums.CouponTaskSendTypeEnum;
import com.nageoffer.onecoupon.merchant.admin.common.enums.CouponTaskStatusEnum;
import com.nageoffer.onecoupon.merchant.admin.dao.entity.CouponTaskDO;
import com.nageoffer.onecoupon.merchant.admin.dao.mapper.CouponTaskMapper;
import com.nageoffer.onecoupon.merchant.admin.dto.req.CouponTaskCreateReqDTO;
import com.nageoffer.onecoupon.merchant.admin.dto.resp.CouponTemplateQueryRespDTO;
import com.nageoffer.onecoupon.merchant.admin.mq.event.CouponTaskExecuteEvent;
import com.nageoffer.onecoupon.merchant.admin.mq.producer.CouponTaskActualExecuteProducer;
import com.nageoffer.onecoupon.merchant.admin.service.CouponTaskService;
import com.nageoffer.onecoupon.merchant.admin.service.CouponTemplateService;
import com.nageoffer.onecoupon.merchant.admin.service.handler.excel.RowCountListener;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.stereotype.Service;

import java.util.Objects;
import java.util.concurrent.*;

/**
 * 优惠券推送业务逻辑实现层
 * <p>
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class CouponTaskServiceImpl extends ServiceImpl<CouponTaskMapper, CouponTaskDO> implements CouponTaskService {

    private final CouponTemplateService couponTemplateService;
    private final CouponTaskMapper couponTaskMapper;
    private final RedissonClient redissonClient;
    private final RocketMQTemplate rocketMQTemplate;
    private final ConfigurableEnvironment configurableEnvironment;
    private final CouponTaskActualExecuteProducer couponTaskActualExecuteProducer;

    /**
     * 自定义线程池
     */
    private final ExecutorService executorService = new ThreadPoolExecutor(
            //corePoolSize核心线程数：因为属于后管任务，大概率不会很频繁，所以直接取服务器CPU核数。
            Runtime.getRuntime().availableProcessors(),
            //maximumPoolSize最大线程数：运行任务属于IO密集型，最大线程数直接服务器CPU核数2倍。
            Runtime.getRuntime().availableProcessors() * 2,
            60L,
            TimeUnit.SECONDS,
            //workQueue任务队列：理论上说我们不会有阻塞的情况，因为设置的线程数不少，所以使用不存储任务的同步队列。
            new SynchronousQueue<>(),
            //handler线程工厂：如果线程数都在运行，直接将任务丢弃即可，因为我们还有延时队列兜底。
            new ThreadPoolExecutor.DiscardPolicy()
    );

    @Override
    public void createCouponTask(CouponTaskCreateReqDTO requestParam) {
        // 验证非空参数
        // 验证参数是否正确，比如文件地址是否为我们期望的格式等
        // 验证参数依赖关系，比如选择定时发送，发送时间是否不为空等
        CouponTemplateQueryRespDTO couponTemplate = couponTemplateService.findCouponTemplateById(requestParam.getCouponTemplateId());
        if (couponTemplate == null) {
            throw new ClientException("优惠券模板不存在，请检查提交信息是否正确");
        }
        // ......

        // 构建优惠券推送任务数据库持久层实体
        CouponTaskDO couponTaskDO = BeanUtil.copyProperties(requestParam, CouponTaskDO.class);
        couponTaskDO.setBatchId(IdUtil.getSnowflakeNextId());
        couponTaskDO.setOperatorId(Long.parseLong(UserContext.getUserId()));
        couponTaskDO.setShopNumber(UserContext.getShopNumber());
        couponTaskDO.setStatus(
                Objects.equals(requestParam.getSendType(), CouponTaskSendTypeEnum.IMMEDIATE.getType())
                        ? CouponTaskStatusEnum.IN_PROGRESS.getStatus()
                        : CouponTaskStatusEnum.PENDING.getStatus()
        );

        // 保存优惠券推送任务记录到数据库
        couponTaskMapper.insert(couponTaskDO);
        log.info("创建优惠券推送任务成功，优惠券推送任务ID：{}", couponTaskDO.getId());
        // 异步执行优惠券推送任务
        JSONObject delayJsonObject = new JSONObject();
        delayJsonObject.put("couponTaskId", couponTaskDO.getId());
        delayJsonObject.put("fileAddress", requestParam.getFileAddress());
        // 线程池执行任务
        executorService.execute(() -> refreshCouponTaskSendNum(delayJsonObject));
        // 使用线程池执行任务的缺点：线程池中的任务还没有运行完服务宕机了，导致任务丢失，所以需要有兜底方案。
        // 可以使用多种方法，比如定时任务，或者使用消息队列，或者使用Redis延迟队列等
        sendDelayMessage(couponTaskDO.getId(), delayJsonObject);
        // 如果是立即发送任务，直接调用消息队列进行发送流程
        if (Objects.equals(requestParam.getSendType(), CouponTaskSendTypeEnum.IMMEDIATE.getType())) {
            CouponTaskExecuteEvent couponTaskExecuteEvent = CouponTaskExecuteEvent.builder()
                    .couponTaskId(couponTaskDO.getId())
                    .build();
            couponTaskActualExecuteProducer.sendMessage(couponTaskExecuteEvent);
        }
    }

    /**
     * 发送延迟任务消息
     * 使用Redisson的阻塞队列和延迟队列来确保任务在一定的延迟后被正确处理
     *
     * @param taskId          优惠券推送任务ID
     * @param delayJsonObject 延迟任务参数
     */
    private void sendDelayMessage(Long taskId, JSONObject delayJsonObject) {
        //阻塞双端队列（RBlockingDeque）：这种类型的队列可以在没有可用元素时阻塞当前线程，适合用于生产者-消费者模型。
        RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque("COUPON_TASK_SEND_NUM_DELAY_QUEUE");
        //通过阻塞双端队列创建了一个延迟队列（RDelayedQueue）：这个延迟队列使得可以将任务放入队列中，并设置一个延迟时间，在这个时间结束后，任务才会被处理。
        RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
        //将任务放入延迟队列：并设置延迟时间为20秒。意思是在20秒之后，delayJsonObject才会被取出并处理。
        // 这里延迟时间设置 20 秒，原因是我们笃定上面线程池20秒之内就能结束任务
        delayedQueue.offer(delayJsonObject, 20, TimeUnit.SECONDS);
        log.info("将优惠券推送任务ID：{}加入延迟队列，延迟时间：20秒", taskId);
    }

    /**
     * 刷新优惠券推送记录中发送行数
     *
     * @param delayJsonObject 延迟任务参数
     */
    private void refreshCouponTaskSendNum(JSONObject delayJsonObject) {
        // 通过 EasyExcel 监听器获取 Excel 中所有行数
        RowCountListener listener = new RowCountListener();
        EasyExcel.read(delayJsonObject.getString("fileAddress"), listener).sheet().doRead();
        // 为什么需要统计行数？因为发送后需要比对所有优惠券是否都已发放到用户账号
        int totalRows = listener.getRowCount();
        // 刷新优惠券推送记录中发送行数
        CouponTaskDO updateCouponTaskDO = CouponTaskDO.builder()
                .id(delayJsonObject.getLong("couponTaskId"))
                .sendNum(totalRows)
                .build();
        couponTaskMapper.updateById(updateCouponTaskDO);
        log.info("刷新优惠券推送记录中发送行数成功，优惠券推送任务ID：{}，发送行数：{}", delayJsonObject.getLong("couponTaskId"), totalRows);
    }

    /**
     * 因为静态内部类的 Bean 注入有问题，所以我们这里直接 new 对象运行即可
     * 如果按照上一版本的方式写，refreshCouponTaskSendNum 方法中 couponTaskMapper 为空
     */
    @PostConstruct
    public void init() {
        new RefreshCouponTaskDelayQueueRunner(this, couponTaskMapper, redissonClient).run();
    }

    /**
     * 优惠券延迟刷新发送条数兜底消费者｜这是兜底策略，一般来说不会执行这段逻辑
     * 如果延迟消息没有持久化成功，或者 Redis 挂了怎么办？后续可以人工处理
     */
    @RequiredArgsConstructor
    static class RefreshCouponTaskDelayQueueRunner {

        private final CouponTaskServiceImpl couponTaskService;
        private final CouponTaskMapper couponTaskMapper;
        private final RedissonClient redissonClient;

        public void run() {
            Executors.newSingleThreadExecutor(
                            runnable -> {
                                Thread thread = new Thread(runnable);
                                thread.setName("delay_coupon-task_send-num_consumer");
                                thread.setDaemon(Boolean.TRUE);
                                return thread;
                            })
                    .execute(() -> {
                        RBlockingDeque<JSONObject> blockingDeque = redissonClient.getBlockingDeque("COUPON_TASK_SEND_NUM_DELAY_QUEUE");
                        log.info("优惠券延迟刷新发送条数兜底消费者启动成功，队列名称：COUPON_TASK_SEND_NUM_DELAY_QUEUE");
                        for (; ; ) {
                            try {
                                // 获取延迟队列已到达时间元素
                                JSONObject delayJsonObject = blockingDeque.take();
                                if (delayJsonObject != null) {
                                    // 获取优惠券推送记录，查看发送条数是否已经有值，有的话代表上面线程池已经处理完成，无需再处理
                                    CouponTaskDO couponTaskDO = couponTaskMapper.selectById(delayJsonObject.getLong("couponTaskId"));
                                    if (couponTaskDO.getSendNum() == null) {
                                        couponTaskService.refreshCouponTaskSendNum(delayJsonObject);
                                        log.info("延迟任务处理成功，优惠券推送任务ID：{}，发送行数：{}", delayJsonObject.getLong("couponTaskId"), couponTaskDO.getSendNum());
                                    }
                                }
                            } catch (Throwable ignored) {
                            }
                        }
                    });
        }
    }
}
